package com.cnlive.logpare.page

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.commons.lang3.time.FastDateFormat
import java.util.Locale
import java.net.URLDecoder
import scala.collection.immutable.List
import scala.util.control.Breaks._
import scala.collection.mutable.HashMap
import java.util.Date
import com.cnlive.logpare.util.PageLog
import com.cnlive.logpare.util.IPSeekerExt
import com.cnlive.logpare.util.ParseLogUtil
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.LongType
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.IntegerType
import java.text.SimpleDateFormat

object ParsePageLog {
  var d=new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss",Locale.ENGLISH)
  val dateFormat = FastDateFormat.getInstance("dd/MMM/yyyy:HH:mm:ss",Locale.ENGLISH)
  val dateFormat2 = FastDateFormat.getInstance("yyyyMMddHHmmss", Locale.ENGLISH)
  val dateFormat3 = FastDateFormat.getInstance("yyyy-MM-dd", Locale.ENGLISH)
  val osList=List("nokia","sony","samsung", "htc", "sharp", "lenovo", "iphone", "ipad", "blackberry", "meizu",
            "android", "symbian", "ucweb", "opera", "cldc", "midp", "wap", "mobile", "htc", "motorola",
            "phone", "xiaomi", "oppo", "vivo", "honor", "huawei", "hisense", "zte", "playstation")
   val pageRe="page.js".r
   val timeRe="0800] \"GET /page.js".r
   val getRepL=".*GET".r
   val headRepL=".*HEAD".r
   def main(args: Array[String]): Unit = {
      var env=1
      if(args.length!=0){
        env=args(0).toInt
      }
      val conf = new SparkConf().setAppName("NewParsePageLog")
      if(env==1){
       conf.setMaster("local")
      }
       val sc = new SparkContext(conf)
       var date=dateFormat2.format(new Date())
       if(args.length!=0){
         date=args(1)
       }
       val files=env match{
         case 1 => "file:///F:/工作资料/page/*"
         case 2 => "hdfs://sxt/spark/test/input/logs/page/*"
         case 3 => //"hdfs://sxt/nginx_logs/events/"+date+"/*";
         "hdfs://sxt/nginx_logs/events/2017-06-06/log-.1496746918082"
      }
      val qqwryPath=env match{
          case 2 =>"/home/yarn/qqwry.dat"
          case 3 =>"/home/dwsj/had/hadoop-2.6.5/lib/qqwry.dat"
          case _=>""
      }
      val zk_addr=env match{
        case 3=>"node001,node002,node003:2181"
        case _=>"node180,node181,node182:2181"
      }
       val person=sc.textFile(files).mapPartitions(x=>{
        var result = List[Row]()
        val iPSeekerExt=env match {
          case 1 =>  IPSeekerExt.getInit()
          case 2 =>  IPSeekerExt.getInit(true,qqwryPath)
          case 3 =>  IPSeekerExt.getInit(true,qqwryPath)
        }
        val sessionMap=new HashMap[String, Int]()
        while (x.hasNext) {
         result= result.::(parLineText(iPSeekerExt,x.next(),sessionMap))
        }
         result.iterator
      }, true)
      .filter(x => x.getInt(0)==1)
      //.count()
      //println(person)
      val sqlContext = new SQLContext(sc)
     // import org.apache.phoenix.spark._
       // Row(0,0L,0L,0L,0L,"","","","","",0,0,0,0,"","",0,"",0)
        val schema = StructType(Seq(
        StructField("isfi", IntegerType, nullable = false),
        StructField("groupTime", LongType, nullable = false),
        StructField("NGINXTIME", LongType, nullable = false),
        StructField("TM", LongType, nullable = false),
        StructField("TMSTR", LongType, nullable = false),
        StructField("UID", StringType, nullable = false),
        StructField("REF", StringType, nullable = false),
        StructField("DOMAIN", StringType, nullable = false),
        StructField("APPID", StringType, nullable = false),
         StructField("IP", StringType, nullable = false),
        StructField("DEVICE", IntegerType, nullable = false),
        StructField("BROWSER", IntegerType, nullable = false),
        StructField("PLAT", IntegerType, nullable = false),
        StructField("COUNTRY", IntegerType, nullable = false),
        StructField("CITY", StringType, nullable = false),
        StructField("LOCATION", StringType, nullable = false),
        StructField("OPERATOR", IntegerType, nullable = false),
        StructField("URI", StringType, nullable = false),
        StructField("ISNEWUSER", IntegerType, nullable = false)
      ))
     val df=  sqlContext.createDataFrame(person, schema)
     val sql= """select 
             NGINXTIME,TM,TMSTR,UID,REF,URI,APPID,
             IP,DEVICE,BROWSER,PLAT,COUNTRY,CITY,
             LOCATION,OPERATOR,DOMAIN
            from t1 """
     df.registerTempTable("t1")
     val dd= sqlContext.sql("""select 
             NGINXTIME,TM,TMSTR,UID,REF,URI,APPID,
             IP,DEVICE,BROWSER,PLAT,COUNTRY,CITY,LOCATION
             ,OPERATOR,ISNEWUSER
            from t1 """)
            //dd.foreach { println(_) }
     // dd.saveToPhoenix("PAGELOG", zkUrl=Some(zk_addr))
      
      
      //
      //sqlContext.createDataFrame(rowRDD, schema)
      //sqlContext.createDataset(person)
       //sqlContext.createDataFrame(person, PageLog.getClass).select("nginxTime").show(100)*/
       
        /*person.saveToPhoenix("PAGELOG",  Seq("NGINXTIME", "TM",
          "TMSTR", "UID",
          "REF", "URI", "APPID", "IP", "DEVICE", "BROWSER", "PLAT", "COUNTRY",
          "CITY",
          "LOCATION","OPERATOR","DOMAIN","ISNEWUSER"),  zkUrl=Some("node180,node182:2181"))*/
   }
   
  def parLineText(iPSeekerExt:IPSeekerExt,lineText:String,sessionMap:HashMap[String, Int]):Row={
         try {
          val text= pageRe.findFirstIn(lineText)
                text match{
                  case Some(lien:String)=>
                   
                     val table=lineText.split("\"").toBuffer
                      println(table)
                     val ip_time=table(0)//220.175.15.117 - - [08/May/2017:10:30:02 +0800]
                     val ip_time_arr=ip_time.split(" ");
                     val ip=ip_time_arr(0)
                     println(ip_time_arr(3).replace("[",""))
                     val time=dateFormat2.format(dateFormat.parse(ip_time_arr(3).replace("[",""))).toLong
                     val statu=table(2)//200 0 
                     val url =table(3).split("\t")(0)//http://sports.china.com.cn/shipin/index.html
                     val browser=if (table.length==6 )URLDecoder.decode(table(5),"utf-8").toLowerCase() else ""
                     val getPer=URLDecoder.decode(headRepL.replaceAllIn(getRepL.replaceAllIn(table(1), ""), ""),"utf-8")//GET /page.js?sver=3.1&uid=dod0k71742_1494210385000&tm=
                     val k=getPer.split("[?]")
                     var uid,ref,uri,appid=""
                     var tm,tmStr=0L
                     var isNewUser=0
                     var browserOs,device,plat=99
                     if(k.length==2){
                         k(1).split("&").map(x=>{
                            val ky=x.split("=")
                            if(ky.length==2){
                              if(ky(0).equals("uid"))uid=ky(1)
                              if(ky(0).equals("tm")){tm=ky(1).toLong;tmStr=dateFormat2.format(ky(1).toLong).toLong}
                              if(ky(0).equals("ref"))ref=ky(1).replaceAll(",", "_")
                              if(ky(0).equals("uri"))uri=ky(1).split(" ")(0).replaceAll(",", "_")
                              if(ky(0).equals("appid"))appid=ky(1)
                            }
                         })
                         if(!uid.equals("")){//uid 
                           val uidAndtm=uid.split("_")
                           if(uidAndtm(0).length()==13){//uid中包含1494210652000_j1tsu24521
                             val r=uidAndtm(0).toLong
                              if(r==tm){
                                 isNewUser=1
                               }
                           }else{ //常规的j1tsu24521_1494210652000
                             val r=uidAndtm(1).toLong
                              if(r==tm){
                               val ss= sessionMap.getOrElse(uid, 0)
                               if(ss==0){
                                 isNewUser=1
                                 sessionMap.put(uid, 1)
                               }
                             }
                           }
                          
                         }
                         if(!browser.equals("")){
                            var tmpbrowser=browser.toLowerCase()
                            if(tmpbrowser.indexOf("mozilla") != -1){
                              tmpbrowser=tmpbrowser.substring(tmpbrowser.indexOf("mozilla")+11)
                            }
                            var flag=true
                            breakable {
                               for(os<-osList){//判断是移动端的是PC端的
                                if(tmpbrowser.indexOf(os) != -1){
                                  flag=false
                                  break
                                }
                               }
                            }
                            if(flag){
                              browserOs=ParseLogUtil.get_pc_browser(tmpbrowser)//PC浏览器类型
                              plat=ParseLogUtil.get_pc_system(tmpbrowser)//PC系统类型
                            }else{
                               device=ParseLogUtil.get_decive(tmpbrowser)//手机型号
                               browserOs=ParseLogUtil.get_mobile_browser(tmpbrowser)//手机浏览器类型
                               plat=ParseLogUtil.get_mobile_system(tmpbrowser)//手机系统
                            }
                         }
                         val d=iPSeekerExt.analyticIp(ip)
                         val city=d.getCity
                         val province=d.getProvince
                         val country=d.getCountry
                         val operator=d.getOperator
                         val groupTime=String.valueOf(time).substring(0, 8).toLong
                         val url_ref=ref
                         val domain=ParseLogUtil.get_domain(uri)
                         Row(1,groupTime,time,tm,tmStr,uid,ref,domain,appid,ip,device,browserOs,plat,country,city,province,operator,url_ref,isNewUser)
                     }else{
                         Row(0,0L,0L,0L,0L,"","","","","",0,0,0,0,"","",0,"",0)
                     }
                  case _=>
                    Row(0,0L,0L,0L,0L,"","","","","",0,0,0,0,"","",0,"",0)
                }
           }catch {
             case ex:Exception=>println(ex.printStackTrace()); Row(3,0L,0L,0L,0L,"","","","","",0,0,0,0,"","",0,"",0)
           }
      }
  
}